package com.mall.promo.mq;

import com.alibaba.fastjson.JSON;
import com.mall.order.dto.CreateSeckillOrderRequest;
import com.mall.promo.cache.CacheManager;
import com.mall.promo.dal.persistence.PromoItemMapper;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.nio.charset.Charset;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.locks.Lock;

@Component
@Slf4j
public class CreatePromoOrderProducer {

    private TransactionMQProducer producer;

    @Autowired
    PromoItemMapper itemMapper;

    @Autowired
    CacheManager cacheManager;

    @Autowired
    RedissonClient client;

    @PostConstruct
    public void init() {
        producer = new TransactionMQProducer("create_promo_order_producer");

        producer.setNamesrvAddr("127.0.0.1:9876");

        // 设置事物消息监听器
        producer.setTransactionListener(new TransactionListener() {
            @Override
            public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
                Map<String, Long> paraMap = (Map<String, Long>) arg;
                Long psId = paraMap.get("psId");
                Long productId = paraMap.get("productId");
                Integer affectRows = 0;

                String lockKey = "promo_stock_lock:" + psId + "-" + productId;
                // 1. 获取锁
                RLock stockLock = client.getLock(lockKey);

                //2. 使用锁,加锁, 如果加锁失败，会阻塞等待
                stockLock.lock();
                try {
                    affectRows = itemMapper.decreaseStock(productId, psId);
                } finally {
                    // 释放锁
                    stockLock.unlock();
                }


                String localTransactionKey = "local_transaction_status:" + msg.getTransactionId();

                String promoProductStockKey = "promo_stock_not_enough: " + psId + "-" + productId;
                if (affectRows < 1) {
                    // 秒杀商品没有库存了,秒杀失败
                    cacheManager.setCache(promoProductStockKey, "yes", 1);
                    cacheManager.setCache(localTransactionKey, "failed", 1);
                    return LocalTransactionState.ROLLBACK_MESSAGE;
                }

                cacheManager.setCache(localTransactionKey, "success", 1);
                return LocalTransactionState.COMMIT_MESSAGE;
            }

            @Override
            public LocalTransactionState checkLocalTransaction(MessageExt msg) {
                String localTransactionKey = "local_transaction_status:" + msg.getTransactionId();

                String transactionStatus = cacheManager.checkCache(localTransactionKey);
                if (transactionStatus == null || transactionStatus.trim().isEmpty()) {
                    // 此时在redis中，没有事物执行的结果
                    return LocalTransactionState.UNKNOW;
                }

                if ("success".equals(transactionStatus.trim())) {
                    return LocalTransactionState.COMMIT_MESSAGE;
                }

                return LocalTransactionState.ROLLBACK_MESSAGE;
            }
        });

        // 启动producer
        try {
            producer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }
    }


    /**
     * 发送事务消息，创建订单
     *
     * @param request
     * @param psId
     * @param productId
     * @return
     */

    public boolean sendPromoOrderMessageInTransaciton(CreateSeckillOrderRequest request, Long psId, Long productId) {
        // 将参数对象转化为json字符串
        String dataStr = JSON.toJSONString(request);

        Map<String, Long> paramMap = new HashMap<>();
        paramMap.put("psId", psId);
        paramMap.put("productId", productId);

        // 准备待发送的消息
        Message message = new Message();
        message.setTopic("create_promo_order");
        message.setBody(dataStr.getBytes(Charset.forName("utf-8")));

        //发送事物消息
        TransactionSendResult sendResult = null;

        try {
            sendResult = producer.sendMessageInTransaction(message, paramMap);
        } catch (MQClientException e) {
            e.printStackTrace();
        }

        if (sendResult != null && LocalTransactionState.COMMIT_MESSAGE.equals(sendResult.getLocalTransactionState())) {
            // 消息发送成功，并且本地事物执行成功
            log.info("秒杀新建订单消息发送成功，并且本地事物执行成功:{}", sendResult);
            log.info("sendResult.getLocalTransactionState()=" + sendResult.getLocalTransactionState());
            return true;
        }
        log.info("秒杀新建订单消息发送失败，或者本地事物执行失败:{}", sendResult);
        return false;
    }
}
